package cn.lecosa.spark.hbase

import org.apache.hadoop.hbase.HBaseConfiguration
import org.apache.hadoop.hbase.client.Put
import org.apache.hadoop.hbase.io.ImmutableBytesWritable
import org.apache.hadoop.hbase.mapred.TableOutputFormat
import org.apache.hadoop.hbase.util.Bytes
import org.apache.hadoop.mapred.JobConf
import org.apache.spark.SparkConf
import org.apache.spark.SparkContext
import org.apache.spark.rdd.RDD.rddToPairRDDFunctions

object WriteHBase {
//将RDD写入hbase
  def main(args: Array[String]): Unit = {
    val sparkConf = new SparkConf().setAppName("HBaseTest").setMaster("local")
    val sc = new SparkContext(sparkConf)

    val hbaseconf = HbaseClient.getHBaseConfiguration;

    val tablename = "trail-data"
    
    //初始化jobconf，TableOutputFormat必须是org.apache.hadoop.hbase.mapred包下的！
    val jobConf = new JobConf(hbaseconf)
    jobConf.setOutputFormat(classOf[TableOutputFormat])
    jobConf.set(TableOutputFormat.OUTPUT_TABLE, tablename)
    
    val indataRDD = sc.makeRDD(Array("1001000018122,460000030171896,null,1514775838,广东 深圳,174,13410716440"))


    
    val rdd=indataRDD.map(_.split(',')).map(save2hbase)
         rdd.foreach(println)
    rdd.saveAsHadoopDataset(jobConf)
    
    sc.stop()
  }

  
  
     def save2hbase (arr:Array[String]) :(ImmutableBytesWritable,Put) = {
      val put = new Put(Bytes.toBytes(arr(0) +arr(1)+arr(3) ) )
      put.add(Bytes.toBytes("cf1"),Bytes.toBytes("DEVICEID"),Bytes.toBytes(arr(0)))
      put.add(Bytes.toBytes("cf1"),Bytes.toBytes("IMSI"),Bytes.toBytes(arr(1)))
      put.add(Bytes.toBytes("cf1"),Bytes.toBytes("IMEI"),Bytes.toBytes(arr(2)))
      put.add(Bytes.toBytes("cf1"),Bytes.toBytes("CAPTURETIME"),Bytes.toBytes(arr(3)))
       put.add(Bytes.toBytes("cf1"),Bytes.toBytes("IMSIADDR"),Bytes.toBytes(arr(4)))
        put.add(Bytes.toBytes("cf1"),Bytes.toBytes("CSQ"),Bytes.toBytes(arr(5)))
         put.add(Bytes.toBytes("cf1"),Bytes.toBytes("SEQ"),Bytes.toBytes(arr(6)))
      //转化成RDD[(ImmutableBytesWritable,Put)]类型才能调用saveAsHadoopDataset
      (new ImmutableBytesWritable, put) 
   }
}